package LearnDataStructure.b_散列结构.应用ConsistentHashing服务器集群;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @version: 1.0
 * @Author: wuqiu
 * @date: 2023-07-06 16:26
 * @description: LeetcodeLearnAlgorithm -> LearnDataStructure.b_散列结构.应用ConsistentHashing服务器集群
 */

/**
 * 进阶实现
 * 考虑数据倾斜
 * 商业成熟方案
 * @param <N>
 */
public class ConsistentHashingPlus<N extends ConsistentHashNodeIMPL> {
    private List<N> nodes = new ArrayList<N>(); // 真实机器节点
    private TreeMap<Long ,N> vNodeAndRealNode;  // 虚拟节点到真实节点的映射
    private TreeMap<Long, N> keyAndRealNode;    // key到真实节点的映射
    private final int V_NODE_NUM = 100;         // 每个机器节点关联的虚拟节点个数
    boolean flag = false;

    private Logger logger = LoggerFactory.getLogger(getClass());

    public ConsistentHashingPlus(List<N> nodes) {
        this.nodes = nodes;
        init();
    }

    /**
     * 初始化函数
     * 初始化一致性hash环
     */
    private void init() {
        vNodeAndRealNode = new TreeMap<Long, N>();
        keyAndRealNode = new TreeMap<Long, N>();
        // 每个真实机器节点都需要关联虚拟节点
        for (int i = 0; i != nodes.size(); ++i) {
            final N shardInfo = nodes.get(i);
            for (int n = 0; n < V_NODE_NUM; n++) {
                // 一个真实机器节点关联NODE_NUM个虚拟节点
                vNodeAndRealNode.put(hash("SHARD-" + shardInfo.getName() + "-NODE-" + n), shardInfo);
            }
        }
    }

    /**
     * 查看当前的请求归属情况
     * 列出所有请求
     */
    public void printKeyTree() {
        logger.debug("当前映射信息为：");
        for (Iterator<Long> it = keyAndRealNode.keySet().iterator(); it.hasNext(); ) {
            Long lo = it.next();
            logger.debug("hash(" + lo + ")连接到主机->" + keyAndRealNode.get(lo));
        }
    }

    /**
     * 添加新机器节点
     * @param s
     */
    public void addNode(N s) {
        logger.debug("增加主机" + s + "的变化：");
        for (int n = 0; n < V_NODE_NUM; n++) {
            addNode(hash("SHARD-" + s.getName() + "-NODE-" + n), s);
        }
    }

    /**
     * 添加新虚拟节点
     * 这里是环添加节点的具体实现
     * @param lg
     * @param s
     */
    private void addNode(Long lg, N s) {
        SortedMap<Long, N> tail = vNodeAndRealNode.tailMap(lg);
        SortedMap<Long, N> head = vNodeAndRealNode.headMap(lg);
        Long begin = 0L;
        Long end = 0L;
        SortedMap<Long, N> between;
        if (head.size() == 0) {
            between = keyAndRealNode.tailMap(vNodeAndRealNode.lastKey());
            flag = true;
        } else {
            begin = head.lastKey();
            between = keyAndRealNode.subMap(begin, lg);
            flag = false;
        }
        vNodeAndRealNode.put(lg, s);
        for (Iterator<Long> it = between.keySet().iterator(); it.hasNext(); ) {
            Long lo = it.next();
            if (flag) {
                keyAndRealNode.put(lo, vNodeAndRealNode.get(lg));
                logger.debug("hash(" + lo + ")改变到->" + tail.get(tail.firstKey()));
            } else {
                keyAndRealNode.put(lo, vNodeAndRealNode.get(lg));
                logger.debug("hash(" + lo + ")改变到->" + tail.get(tail.firstKey()));
            }
        }
    }

    /**
     * 删除机器节点
     * @param s
     */
    public void deleteNode(N s) {
        if (s == null) {
            return;
        }
        logger.debug("删除主机" + s + "的变化：");
        for (int i = 0; i < V_NODE_NUM; i++) {
            //定位s节点的第i的虚拟节点的位置
            SortedMap<Long, N> tail = vNodeAndRealNode.tailMap(hash("SHARD-" + s.getName() + "-NODE-" + i));
            SortedMap<Long, N> head = vNodeAndRealNode.headMap(hash("SHARD-" + s.getName() + "-NODE-" + i));
            Long begin = 0L;
            Long end = 0L;

            SortedMap<Long, N> between;
            if (head.size() == 0) {
                between = keyAndRealNode.tailMap(vNodeAndRealNode.lastKey());
                end = tail.firstKey();
                tail.remove(tail.firstKey());
                vNodeAndRealNode.remove(tail.firstKey());// 从nodes中删除s节点的第i个虚拟节点
                flag = true;
            } else {
                begin = head.lastKey();
                end = tail.firstKey();
                tail.remove(tail.firstKey());
                between = keyAndRealNode.subMap(begin, end);// 在s节点的第i个虚拟节点的所有key的集合
                flag = false;
            }
            for (Iterator<Long> it = between.keySet().iterator(); it.hasNext(); ) {
                Long lo = it.next();
                if (flag) {
                    keyAndRealNode.put(lo, tail.get(tail.firstKey()));
                    logger.debug("hash(" + lo + ")改变到->" + tail.get(tail.firstKey()));
                } else {
                    keyAndRealNode.put(lo, tail.get(tail.firstKey()));
                    logger.debug("hash(" + lo + ")改变到->" + tail.get(tail.firstKey()));
                }
            }
        }
    }

    /**
     * 映射key到机器节点
     * 也就是映射新的请求到真实的节点
     * @param key
     */
    public void keyToNode(String key) {
        SortedMap<Long, N> tail = vNodeAndRealNode.tailMap(hash(key)); // 沿环的顺时针找到一个虚拟节点
        if (tail.size() == 0) {
            return;
        }
        Long virtualNodeKey = tail.firstKey();
        N realNode = tail.get(virtualNodeKey);
        keyAndRealNode.put(hash(key), realNode);
        logger.debug(key + "（hash：" + hash(key) + "）连接到主机->" + realNode);
    }

    /**
     * MurMurHash算法，是非加密HASH算法，性能很高，
     * 比传统的CRC32,MD5，SHA-1（这两个算法都是加密HASH算法，复杂度本身就很高，带来的性能上的损害也不可避免）
     * 等HASH算法要快很多，而且据说这个算法的碰撞率很低.
     * http://murmurhash.googlepages.com/
     *
     * @param key
     * @return
     */
    private static Long hash(String key) {
        ByteBuffer buf = ByteBuffer.wrap(key.getBytes());
        final int seed = 0x1234ABCD;
        final int r = 47;
        final long m = 0xc6a4a7935bd1e995L;
        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);
        long h = seed ^ (buf.remaining() * m);
        long k;
        while (buf.remaining() >= 8) {
            k = buf.getLong();

            k *= m;
            k ^= k >>> r;
            k *= m;

            h ^= k;
            h *= m;
        }
        if (buf.remaining() > 0) {
            ByteBuffer finish = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
            // for big-endian version, do this first:
            // finish.position(8-buf.remaining());
            finish.put(buf).rewind();
            h ^= finish.getLong();
            h *= m;
        }
        h ^= h >>> r;
        h *= m;
        h ^= h >>> r;
        buf.order(byteOrder);
        return Math.abs(h);
    }
}